package com.example.emos.wx.task;

import com.example.emos.wx.db.pojo.MessageEntity;
import com.example.emos.wx.db.pojo.MessageRefEntity;
import com.example.emos.wx.exception.EmosException;
import com.example.emos.wx.service.MessageService;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

@Component
@Slf4j
public class MessageTask {

    @Autowired
    private ConnectionFactory factory;

    @Autowired
    private MessageService messageService;

    /**
     * 发送数据
     * @param topic 每一个队列都有自己的名字就叫做topic
     * @param entity
     */
    public void send(String topic, MessageEntity entity){
        String id = messageService.insertMessage(entity);
        //创建连接
        try(Connection connection = factory.newConnection();
            //创建通道
            Channel channel = connection.createChannel();
        ){
            /**
             * 五个参数分别为
             * 要连接的队列（没有的话会自动创建）
             * 是否持久化
             * 是否加锁
             * 是否自动删除
             * 一些额外参数
             */
            channel.queueDeclare(topic,true,false,false, null);
            HashMap map = new HashMap();
            map.put("messageId",id);
            //创建AMQP对象
            AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().headers(map).build();
            //发送消息
            channel.basicPublish("",topic,properties,entity.getMsg().getBytes());
            log.debug("消息发送成功");
        }catch (Exception e){
            log.error("执行异常",e);
            throw new EmosException("向MQ发送消息失败");
        }
    }

    @Async
    public void sendAsync(String topic, MessageEntity entity){
        send(topic,entity);
    }


    /**
     * 接收数据
     * @param topic 从那个队列接收数据
     * @return
     */
    public int receive(String topic){
        //用来记录收到消息的数量
        int i = 0;
        try(Connection connection = factory.newConnection();
            //创建通道
            Channel channel = connection.createChannel();
        ){
            channel.queueDeclare(topic,true,false,false, null);
            while (true){
                /**
                 *
                 * 目标队列
                 * 是否自动发送Ack应答
                 */
                GetResponse response = channel.basicGet(topic, false);
                if (response != null){
                    AMQP.BasicProperties properties = response.getProps();
                    Map<String, Object> map = properties.getHeaders();
                    String messageId = map.get("messageId").toString();
                    byte[] body = response.getBody();
                    String message = new String(body);
                    log.debug("从RabbitMQ接收的消息：" + message);
                    MessageRefEntity entity = new MessageRefEntity();
                    entity.setMessageId(messageId);
                    entity.setReceiverId(Integer.parseInt(topic));
                    entity.setReadFlag(false);
                    entity.setLastFlag(true);
                    messageService.insertRef(entity);

                    long deliveryTag = response.getEnvelope().getDeliveryTag();
                    //返回Ack应答
                    channel.basicAck(deliveryTag, false);
                    i++;
                }
                else {
                    break;
                }
            }


        }catch (Exception e){
            log.error("执行异常",e);
            throw new EmosException("接收消息失败");
        }

        return i;
    }

    @Async
    public int receiveAsync(String topic){
        return receive(topic);
    }

    /**
     * 删除消息队列
     * @param topic
     */
    public void deleteQueue(String topic){
        //创建连接
        try(Connection connection = factory.newConnection();
            //创建通道
            Channel channel = connection.createChannel();
        ){
            channel.queueDelete(topic);
            log.debug("消息队列删除成功");
        }catch (Exception e){
            log.error("删除队列失败",e);
            throw new EmosException("删除队列失败");
        }
    }

    @Async
    public void deleteQueueAsync(String topic){
        deleteQueue(topic);
    }


}
